src/write-queue.ts (5)

14-14: Consider accepting both single and multiple tasks in one property.
While this approach is valid, you may explore a type union for the property (e.g., task?: DocUpdate | DocUpdate[] ) to unify the handling.

---

26-32: Check for valid configuration in defaultWriteQueueOpts.
Consider validating that opts.chunkSize > 0 or throwing an error for invalid input, instead of silently accepting.

---

34-47: Potential constructor safeguard.
You might validate or sanitize opts before assignment, ensuring chunkSize is within a safe range. Also consider logging if an unexpected value is received.

---

78-82: Bulk method design is coherent.
The logic is aligned with the new queue structure. Consider adding optional concurrency controls if large bulks can overwhelm downstream processes.

---

84-89: Push method is consistent with bulk.
Both methods unify well under the same queue-based approach. Might consider factoring out shared logic to avoid duplication if it grows more complex.

tests/fireproof/ledger.test.ts (3)

236-268: Consider adding error scenario test cases.

The test suite verifies the happy path for ordered writes, but it would be beneficial to add test cases for error scenarios, such as:

• Network failures during writes
• Concurrent writes with conflicts
• Invalid document formats

it("should handle write failures gracefully", async () => {
  // Mock a network failure
  const failingDoc = { _id: "fail-test", hello: "world" };
  // ... test implementation
});

---

329-330: Remove commented console.log statement.

Clean up the test code by removing the commented debug statement.

rows.sort((a, b) => a.key.localeCompare(b.key));
-// console.log(rows);
for (let i = 0; i < 10; i++) {

---

Line range hint 236-330: Consider implementing a shared test factory for write queue configurations.

The two test suites share similar setup and verification logic but with different chunk sizes. Consider creating a shared test factory that can run the same test cases with different configurations to reduce code duplication and make it easier to test additional configurations in the future.

Example approach:

const testWriteQueue = (description: string, config: { chunkSize: number }) => {
  describe(`basic Ledger parallel writes / ${description}`, () => {
    // Shared test setup and cases
  });
};

testWriteQueue("public ordered", { chunkSize: 1 });
testWriteQueue("public parallel", { chunkSize: 32 });

Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro

src/write-queue.ts (1)

55-76: Consider enhancing error handling and process scheduling.

Two potential improvements for the process method:

1. Consider adding batch-level error handling to catch and log any unexpected errors during processing.
2. Replace the recursive void this.process() call with a non-recursive approach to prevent potential stack overflow with large queues.

-    void this.process();
+    setImmediate(() => void this.process());

Configuration used: .coderabbit.yaml
Review profile: CHILL
Plan: Pro

tests/fireproof/crdt.test.ts (1)

466-466: Applying default queue in Loader with many transactions.

Ensures concurrency patterns remain uniform in large-scale commit tests. Keep an eye on potential throughput or memory constraints when scaling.

Configuration used: .coderabbit.yaml
Review profile: CHILL
Plan: Pro

#!/bin/bash
# Search for defaultWriteQueueOpts definition and usage
rg "defaultWriteQueueOpts" -A 5

# Search for writeQueue configurations in test files
rg "writeQueue:" --type ts -A 3 tests/

#!/bin/bash
# Search for writeQueue usage in test files
rg "writeQueue" tests/ -B 2 -A 2

# Search for compaction related tests and their write queue configuration
rg "compaction|compact" tests/ -B 2 -A 2

# Look for defaultWriteQueueOpts definition and usage
ast-grep --pattern 'defaultWriteQueueOpts'

#!/bin/bash
# Search for defaultWriteQueueOpts definition and usage
rg "defaultWriteQueueOpts" -A 5

# Also search for writeQueue configurations in test files
rg "writeQueue" tests/ -A 3

#!/bin/bash
# Look for WriteQueueParams interface and related configurations
cat src/write-queue.ts

#!/bin/bash
# Search for write queue related tests and implementations
rg -l "writeQueue" --type ts

# Look for loader test cases
rg "loader.*test" --type ts

# Search for defaultWriteQueueOpts usage
rg "defaultWriteQueueOpts" -A 5 --type ts

#!/bin/bash
# Get the loader test implementation and surrounding context
rg -B 5 -A 20 "test-loader" tests/fireproof/crdt.test.ts

# Check write queue implementation to understand default behavior
cat src/write-queue.ts

src/write-queue.ts (4)

20-24: Enhance WriteQueueParams documentation.

The interface documentation should clarify:

• The purpose and impact of chunkSize
• Valid range of values
• Performance implications

 export interface WriteQueueParams {
-  // default 32
-  // if chunkSize is 1 the result will be ordered in time
+  /**
+   * Maximum number of tasks to process in a single batch.
+   * - Default: 32
+   * - Set to 1 for strict time-ordered processing
+   * - Higher values improve throughput but may affect ordering
+   * @minimum 1
+   */
   readonly chunkSize: number;
 }

---

14-15: Consider using union type for tasks as suggested.

The current implementation uses an optional tasks array, but a union type might be clearer as suggested in the previous review.

 interface WriteQueueItem<T extends DocTypes> {
-  // readonly task?: DocUpdate<T>;
-  readonly tasks?: DocUpdate<T>[];
+  readonly task: DocUpdate<T> | DocUpdate<T>[];
   resolve(result: MetaType): void;
   reject(error: Error): void;
 }

---

55-76: Enhance process method robustness.

The current implementation has several areas for improvement:

1. No timeout handling for worker operations
2. Memory usage could grow unbounded without queue size limits
3. Error handling could be more informative

Consider implementing:

1. Worker timeout mechanism
2. Queue size limits
3. Detailed error reporting

Example implementation:

private readonly MAX_QUEUE_SIZE = 10000;
private readonly WORKER_TIMEOUT = 30000; // 30 seconds

private async process() {
  if (this.isProcessing || this.queue.length === 0) {
    this.testEmptyQueue();
    return;
  }
  
  if (this.queue.length > this.MAX_QUEUE_SIZE) {
    this.logger.Warn().Int("queueSize", this.queue.length).Msg("Queue size exceeds limit");
  }

  this.isProcessing = true;
  const tasksToProcess = this.queue.splice(0, this.opts.chunkSize);
  
  const processWithTimeout = async (update: DocUpdate<T>[], index: number) => {
    const timeoutPromise = new Promise((_, reject) => {
      setTimeout(() => reject(new Error("Worker timeout")), this.WORKER_TIMEOUT);
    });
    
    try {
      const result = await Promise.race([this.worker(update), timeoutPromise]);
      tasksToProcess[index].resolve(result);
    } catch (error) {
      const enhancedError = new Error(
        `Task processing failed: ${error.message}`,
        { cause: error }
      );
      tasksToProcess[index].reject(enhancedError);
    }
  };

  // ... rest of the implementation
}

---

78-94: Consider adding queue management features.

The current implementation could benefit from additional queue management capabilities:

• Queue status monitoring
• Ability to pause/resume processing
• Queue statistics

Example additions:

interface WriteQueue<T extends DocTypes> {
  // ... existing methods ...
  
  /** Get current queue status */
  status(): {
    length: number;
    isProcessing: boolean;
    isPaused: boolean;
  };
  
  /** Pause processing */
  pause(): void;
  
  /** Resume processing */
  resume(): void;
}

Configuration used: .coderabbit.yaml
Review profile: CHILL
Plan: Pro

src/write-queue.ts (3)

14-14: Remove commented code.

Remove the commented line as it's no longer needed and can cause confusion.

-  // readonly task?: DocUpdate<T>;

---

26-31: Consider adding upper bound validation and documentation.

While the implementation handles the lower bound well, consider these improvements:

1. Add an upper bound for chunkSize to prevent potential memory issues with very large chunks
2. Document why 32 was chosen as the default value

 export function defaultWriteQueueOpts(opts: Partial<WriteQueueParams> = {}): WriteQueueParams {
+  const DEFAULT_CHUNK_SIZE = 32;
+  const MAX_CHUNK_SIZE = 1000; // or another appropriate limit
+  
+  let chunkSize = opts.chunkSize || DEFAULT_CHUNK_SIZE;
+  if (chunkSize <= 0) {
+    chunkSize = DEFAULT_CHUNK_SIZE;
+  } else if (chunkSize > MAX_CHUNK_SIZE) {
+    chunkSize = MAX_CHUNK_SIZE;
+  }
+  
   return {
     ...opts,
-    chunkSize: opts.chunkSize && opts.chunkSize > 0 ? opts.chunkSize : 32,
+    chunkSize,
   };
 }

---

86-90: Improve cleanup in close method.

The close method should clear any pending tasks to prevent memory leaks.

 close(): Promise<void> {
   this.waitForEmptyQueue = new Future();
   this.testEmptyQueue();
+  // Clear any pending tasks to prevent memory leaks
+  this.queue.length = 0;
   return this.waitForEmptyQueue.asPromise();
 }

Configuration used: .coderabbit.yaml
Review profile: CHILL
Plan: Pro

#!/bin/bash
# Description: Verify the integration of bulk operations across the codebase

# Check for bulk operation usage patterns
echo "Checking bulk operation usage patterns..."
rg -t typescript "bulk\(" --context 3

# Check for WriteQueueParams configuration
echo "Checking WriteQueueParams configuration..."
rg -t typescript "WriteQueueParams" --context 2

# Check for potential race conditions in concurrent bulk operations
echo "Checking for potential race conditions..."
ast-grep --pattern 'bulk($tasks) {
  $$$
  this.queue.push($$$)
  $$$
}'

#!/bin/bash
# Let's try without typescript file type specification
echo "Checking bulk operation usage patterns..."
rg "bulk\(" -A 3

echo "Checking WriteQueueParams configuration..."
rg "WriteQueueParams"

echo "Checking write queue usage..."
rg "writeQueue\(" -A 3

# Check for any potential concurrent write operations
echo "Checking concurrent write patterns..."
rg "push\(" -A 3